package com.shujia.traffic.dws

import com.shujia.traffic.common.WfkjUtils
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

import scala.collection.mutable

object DwsAppVioZqzfDay {
  def main(args: Array[String]): Unit = {
    val bizdate: String = "20250111"

    // 创建SparkSession
    // 如果不在本地执行，则master及config配置在提交任务时统一指定，代码中不写死
    val spark: SparkSession = SparkSession
      .builder()
      .appName("DwsAppVioWfjlgdDay")
      .enableHiveSupport() // 开启Hive的支持，使Spark代码能够获取Hive的表及数据
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 非现场违法表
    val surveilDF: DataFrame = spark.table("dwd.dwd_base_vio_surveil_msk_d_f")
      .where($"pt" === bizdate)
      .select(date_format(coalesce($"wfsj", $"wfsj1"), "yyyy-MM-dd") as "wfsj", $"wfxw")
    // 现场违法表
    val forceDF: DataFrame = spark.table("dwd.dwd_base_vio_force_msk_d_f")
      .where($"pt" === bizdate)
      .select(
        date_format($"wfsj", "yyyy-MM-dd") as "wfsj"
        , explode(array($"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")) as "wfxw"
      )
    // 事故表及事故人员表
    val acdDF: DataFrame = spark.table("dwd.dwd_base_acd_file_msk_d_f").where($"pt" === bizdate)
      .select($"sgfssj", $"sgbh")
    val acdHumanDF: DataFrame = spark.table("dim.dim_base_acd_filehuman_msk_d_f").where($"pt" === bizdate)
      .select($"sgbh", $"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")

    val unionDF: Dataset[Row] = acdDF
      .join(acdHumanDF, List("sgbh"), "inner")
      .select(
        date_format($"sgfssj", "yyyy-MM-dd") as "wfsj"
        , explode(array($"wfxw1", $"wfxw2", $"wfxw3", $"wfxw4", $"wfxw5")) as "wfxw"
      )
      .union(surveilDF)
      .union(forceDF)
      .where($"wfxw".isNotNull and trim($"wfxw") =!= "" and upper(trim($"wfxw")) =!= "NULL" and $"wfsj".isNotNull)

    val wfkjMap: mutable.Map[String, List[String]] = WfkjUtils.wfkjMap

    var drWfsDF: DataFrame = unionDF
      .groupBy($"wfsj")
      .agg(
        count("*") as "dr_zd_wfs" // 当日总的违法数
        // 基于违法行为代码分别统计当日不同类别的违法行为数量
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("zjjs", Nil)), 1).otherwise(0)) as "dr_zjjs_wfs" // 当日zjjs驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("yjjs", Nil)), 1).otherwise(0)) as "dr_yjjs_wfs" // 当日yjjs驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("zdhp", Nil)), 1).otherwise(0)) as "dr_zdhp_wfs" // 当日zdhp驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wzbz", Nil)), 1).otherwise(0)) as "dr_wzbz_wfs" // 当日wzbz驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("tpjp", Nil)), 1).otherwise(0)) as "dr_tpjp_wfs" // 当日tpjp驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wdtk", Nil)), 1).otherwise(0)) as "dr_wdtk_wfs" // 当日wdtk驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("jspz", Nil)), 1).otherwise(0)) as "dr_jspz_wfs" // 当日jspz驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("wxjs", Nil)), 1).otherwise(0)) as "dr_wxjs_wfs" // 当日wxjs驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("cy", Nil)), 1).otherwise(0)) as "dr_cy_wfs" // 当日cy驾驶违法数
        , sum(when($"wfxw".isInCollection(wfkjMap.getOrElse("cz", Nil)), 1).otherwise(0)) as "dr_cz_wfs" // 当日cz驾驶违法数
      )
    val wflxList: List[String] = List("zd", "zjjs", "yjjs", "zdhp", "wzbz", "tpjp", "wdtk", "jspz", "wxjs", "cy", "cz")

    wflxList.foreach(wflx=>{
      drWfsDF = drWfsDF
        .withColumn(s"jn_${wflx}_wfs", sum($"dr_${wflx}_wfs") over Window.partitionBy(year($"wfsj")))
        .withColumn(s"last_dr_${wflx}_wfs", lag($"dr_${wflx}_wfs", 1) over Window.partitionBy(date_format($"wfsj", "MM-dd")).orderBy(year($"wfsj")))
        .withColumn(s"${wflx}_tb", coalesce(round($"dr_${wflx}_wfs" / $"last_dr_${wflx}_wfs" - 1, 4), expr("0")))
    })

    spark.sql(
      s"""
         |alter table dws.dws_app_vio_zqzf_d add if not exists partition(pt='$bizdate')
         |""".stripMargin)

    drWfsDF
      .select(
        $"wfsj"
        ,$"dr_zd_wfs",$"jn_zd_wfs",$"zd_tb"
        ,$"dr_zjjs_wfs",$"jn_zjjs_wfs",$"zjjs_tb"
        ,$"dr_yjjs_wfs",$"jn_yjjs_wfs",$"yjjs_tb"
        ,$"dr_zdhp_wfs",$"jn_zdhp_wfs",$"zdhp_tb"
        ,$"dr_wzbz_wfs",$"jn_wzbz_wfs",$"wzbz_tb"
        ,$"dr_tpjp_wfs",$"jn_tpjp_wfs",$"tpjp_tb"
        ,$"dr_wdtk_wfs",$"jn_wdtk_wfs",$"wdtk_tb"
        ,$"dr_jspz_wfs",$"jn_jspz_wfs",$"jspz_tb"
        ,$"dr_wxjs_wfs",$"jn_wxjs_wfs",$"wxjs_tb"
        ,$"dr_cy_wfs",$"jn_cy_wfs",$"cy_tb"
        ,$"dr_cz_wfs",$"jn_cz_wfs",$"cz_tb"
      )
      .write
      .mode(SaveMode.Overwrite)
      .format("orc")
      .save(s"/project/data/traffic/dws/dws_app_vio_zqzf_d/pt=$bizdate")



  }

}
